import json
from confluent_kafka import Producer
from datetime import datetime

# 配置 Kafka 生产者
conf = {
    'bootstrap.servers': '192.168.141.77:9092',  # Kafka broker 地址
    'client.id': 'producer'
}

producer = Producer(conf)

# 发送消息的回调函数
def delivery_report(err, msg):
    if err is not None:
        print('Message delivery failed: {}'.format(err))
    else:
        print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))

# 获取当前时间并格式化
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

# 发送消息
topic = 'EventRealTimeTarget'
message = {
    "event_id": "5ca42836-5c3f-499f-8f95-08144b657413",
    "from_device": "172.16.10.44_V48",
    "picture_path": "https://pic.rmb.bdstatic.com/bjh/news/121f8fd7af55c6e6415642ef283d2404.jpeg",
    "video_path": "http://vjs.zencdn.net/v/oceans.mp4",
    "event_type": 1006,
    "time": current_time
}

# 将字典转换为 JSON 字符串并编码为 UTF-8
message_json = json.dumps(message)

# 发送消息
producer.produce(topic, message_json.encode('utf-8'), callback=delivery_report)

# 等待所有消息都发送完毕
producer.flush()
